package com.pw.study.flink.source

import com.alibaba.fastjson.JSONObject
import org.apache.flink.streaming.api.functions.source.{RichSourceFunction, SourceFunction}

import scala.util.Random


/**
 * @Desc: 学生数据源
 * @author: future
 * @since: 2022/6/3-11:06
 */
class StudentSource extends RichSourceFunction[JSONObject] {
  private var isSwitch: Boolean = true

  override def run(ctx: SourceFunction.SourceContext[JSONObject]): Unit = {

    val str = "How about one year olde My bitthday past a couple days and there is something what I thought Among the main reasons for delaying"
    val line: Array[String] = str.split(" ")
    val student = "{\"student\":\"lisi\"}"
    val random = new Random()
    while (isSwitch) {
      val obj = new JSONObject()
      val num: Int = random.nextInt(1000 * 60 * 5)

      val len: Int = random.nextInt(line.length)
      val ts: Long = System.currentTimeMillis() + num - 1000 * 60 * 60 * 6
      obj.put("student", "User-" + line(len))
      obj.put("ts", ts)
      obj.put("age", len)

      ctx.collect(obj)
      Thread.sleep(100)
    }

  }

  override def cancel(): Unit = {
    isSwitch = false
  }
}
